package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentInfo;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.utils.DateTimeUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Author: tiancy
 * Date: 2021/12/3
 * Desc: 支付宽表
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、BaseDBApp、OrderWideApp、PaymentWideApp
 */
public class PaymentWideApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置(略)
        //TODO 3.从kafka中读取数据
        //3.1 声明消费的主题以及消费者组
        String paymentInfoTopic = "dwd_payment_info";
        String orderWideTopic = "dwm_order_wide";
        String groupId = "payment_wide_app_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> paymentInfoKafkaSource = MyKafkaUtil.getKafkaSource(paymentInfoTopic, groupId);
        FlinkKafkaConsumer<String> orderWideKafkaSource = MyKafkaUtil.getKafkaSource(orderWideTopic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> paymentInfoStrDS = env.addSource(paymentInfoKafkaSource);
        DataStreamSource<String> orderWideStrDS = env.addSource(orderWideKafkaSource);

        //TODO 4.对读取的数据进行类型的转换  jsonStr--->实体类型对象
        //4.1 支付
        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS = paymentInfoStrDS.map(jsonStr -> JSON.parseObject(jsonStr, PaymentInfo.class));

        //4.2 订单宽表
        SingleOutputStreamOperator<OrderWide> orderWideDS = orderWideStrDS.map(jsonStr -> JSON.parseObject(jsonStr, OrderWide.class));

        //paymentInfoDS.print(">>>");
        //orderWideDS.print("####");

        //TODO 5.指定Watermark以及提取事件时间字段
        //5.1 支付
        SingleOutputStreamOperator<PaymentInfo> paymentInfoWithWatermarkDS = paymentInfoDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<PaymentInfo>() {
                        @Override
                        public long extractTimestamp(PaymentInfo paymentInfo, long recordTimestamp) {
                            return DateTimeUtil.toTs(paymentInfo.getCallback_time());
                        }
                    }
                )
        );

        //5.2 订单宽表
        SingleOutputStreamOperator<OrderWide> orderWideWithWatermarkDS = orderWideDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<OrderWide>() {
                        @Override
                        public long extractTimestamp(OrderWide orderWide, long recordTimestamp) {
                            return DateTimeUtil.toTs(orderWide.getCreate_time());
                        }
                    }
                )
        );

        //TODO 6.通过keyby分组  指定连接字段
        //6.1 支付
        KeyedStream<PaymentInfo, Long> paymentInfoKeyedDS = paymentInfoWithWatermarkDS.keyBy(PaymentInfo::getOrder_id);

        //6.2 订单宽表
        KeyedStream<OrderWide, Long> orderWideKeyedDS = orderWideWithWatermarkDS.keyBy(OrderWide::getOrder_id);

        //TODO 7.使用intervalJoin完成支付和订单宽表的双流join
        SingleOutputStreamOperator<PaymentWide> paymentWideDS = paymentInfoKeyedDS
            .intervalJoin(orderWideKeyedDS)
            .between(Time.seconds(-1800), Time.seconds(0))
            .process(
                new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWide orderWide, Context ctx, Collector<PaymentWide> out) throws Exception {
                        out.collect(new PaymentWide(paymentInfo, orderWide));
                    }
                }
            );

        paymentWideDS.print(">>>>>");

        //TOOD 8.将支付宽表数据写到kafka的dwm层
        paymentWideDS
            .map(paymentInfo->JSON.toJSONString(paymentInfo))
            .addSink(MyKafkaUtil.getKafkaSink("dwm_payment_wide"));

        env.execute();
    }
}
